package com.atguigu.udmp.stream.function;

import com.atguigu.udmp.stream.bean.DimInfo;
import com.atguigu.udmp.stream.common.util.DorisJdbcUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;

import java.util.Set;

@Slf4j
public class DimCheckPartitionFunction extends RichMapFunction<DimInfo,DimInfo> {
    Set<String> partitionSet=null;

    public void open(Configuration parameters)   {
        // 加载当前表的已有分区

        partitionSet= DorisJdbcUtil.showPartition("dim_info");
        log.info("加载已有分区："+partitionSet);
    }

    @Override
    public DimInfo map(DimInfo dimInfo)  {
        Long dimDefineId = dimInfo.getDimDefineId();
        String partitionName="p_"+dimDefineId;   //分区名称
        String[]  partitionValues=new String[]{ dimDefineId+""};
        if(!partitionSet.contains(partitionName)){//不含分区 则添加分区
            DorisJdbcUtil.addPartition("dim_info",partitionName,partitionValues);
            partitionSet.add(partitionName);
        }
        return dimInfo;
    }
}
